/*
   Copyright 2010 Bryce Alcock

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.com

 */


package com.ecocrypt.proxy.forwarder;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Date;
import java.util.LinkedList;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 * @author Bryce.Alcock
 */
public class ClientServiceRequest implements Runnable, Serializable {

    // Only start 1 single thread for all ClientServiceRequest Updates.
    // Each ClientServiceRequest is responsible for ensuring that it cancels itself when
    // there will be no more update.
    private static final Timer dataChangeTimer = new Timer("ClientServiceRequest Notification Timer");


    //  This is used for hooking up publish-subscribe listeners,
    //  and notifying them of data changes.
    //  Data change notification are sent by a timer task if and
    //  only if this attribute is set to true.
    transient boolean dirty = true;
    transient boolean finalTimer = false;


    transient boolean isAlive = true;

    transient Socket clientSocket;
    transient InputStream clientInputStream;
    transient OutputStream clientOutputStream;


    private long startTime;
    private long endTime =-1;
    Forwarder forwarder;
    EngineSocket engineSocket;
    private long lastByteSent =1;



    Integer serviceId;

    int clientLocalPort;
    AtomicInteger bytesFromClient = new AtomicInteger(0);
    AtomicInteger bytesToClient = new AtomicInteger(0);

    //--------------------------------------------------------------------------
    //
    //   MEMORY LEAK POTENTIAL:
    //   The following pieces of data could cause a memory leak, and therefore
    //   Must be dealt with if the size of the data grows.
    //
    //--------------------------------------------------------------------------

    StringBuilder sentData = new StringBuilder();
    StringBuilder recieveData = new StringBuilder();
    LinkedBlockingQueue<TimingPill> sendTimes = new LinkedBlockingQueue<TimingPill>(500);
    LinkedBlockingQueue<TimingPill> recieveTimes = new LinkedBlockingQueue<TimingPill>(500);
    //
    //  End Memory Leak Potential Elements.
    //--------------------------------------------------------------------------

    LinkedList<String> shutdownReason = new LinkedList<String>();

    /**
     * This constructor will assign this request a serviceId, then
     * it will attempt to establish a connection to the engine.  Once that is done,
     * it will initialize the clientInputStreams and clientOutputStreams from the
     * socket.  If all is successful, it will assign the forwarder to this object's
     * forwarder attribute.
     *
     * @param forwarder - the forwarder associated with this request.
     * @param clientSocket - the bound duplex socket back to the client.
     * @param sid - the requestId for this request.
     * @throws IOException - if there is an exception opening the input streams or output streams,
     * or if there is an error creating the EngineSocket.
     */
    public ClientServiceRequest(Forwarder f, Socket s, Integer sid) throws IOException {
        //  1.  Create The Socket to the Engine.
        //  2.  Set All other Attributes of this socket.
        serviceId =sid;

	startTime=System.currentTimeMillis();
        // Establish a socket connection to the Engine.
        //     This will also open up the input and output socket streams
        //     to and from the engine.
        engineSocket = new EngineSocket(f, serviceId, this);

        // Open up the input and output streams from the client.
        clientSocket =s;
        clientSocket.setKeepAlive(ForwardProxy.clientKeepAlive);
        clientInputStream = clientSocket.getInputStream();
        clientOutputStream = clientSocket.getOutputStream();
        InetSocketAddress sa = (InetSocketAddress) s.getRemoteSocketAddress();
        clientLocalPort = sa.getPort();

        //  Assigning for Forward Reference Only.
        forwarder =f;

	TimerTask dataChangedTimerTask= new TimerTask(){

	    @Override
	    public void run() {

		if(finalTimer){
		    this.cancel();
		}

		if(dirty){
		    forwarder.connectionDataChanged(ClientServiceRequest.this);
		    dirty=false;
		}
	    }
	};

	// Is most likely rediculously long, but it can be tuned in the future.
	// Connections should not live more that 1 or 2 seconds at the most.
	// for conservative reasons, a long time was choosen.
	// TODO: Make this tuneable at startup and runtime.
	dataChangeTimer.scheduleAtFixedRate(dataChangedTimerTask, 1000, 5000);

    }

    /**
     * This method should be called only 1 time.
     * This method will start all 4 threads associated with proxying of this
     * full duplex socket.
     */
    public void startServiceThreads(){
	final String clientInputThreadFormatString =
		"[%d]:[%d] Client Input Thread %s";
	final String clientOutputThreadFormatString =
		"[%d]:[%d] Client Output Thread %s";

        engineSocket.startServiceThreads();

	ClientSocketOutputSender senderThread = new ClientSocketOutputSender();
	senderThread.setName(
		String.format(
		clientOutputThreadFormatString, serviceId,clientSocket.getPort(),forwarder.toString()));
        senderThread.start();
	
	Thread clientInputThread = new Thread(this);
	clientInputThread.setName(
		String.format(
		clientInputThreadFormatString, serviceId, clientSocket.getPort() ,forwarder.toString()));
        clientInputThread.start();
    }


    /**
     * This class/thread has the responsibility of reading data from the
     * engineSocket Receive buffer, and writing that data out to the clientOutputStream.
     * This class/thread will continuously poll the stream until the engine's corresponding
     * duplex socket sends a '-1' byte marking the end of the response.  This socket will
     * then "close".
     */
    private class ClientSocketOutputSender extends Thread {

        @Override
        public void run() {
            while (isAlive) {
                if(engineSocket.getRecieveBufferSize()>0){
                    try {
                        LinkedList<Byte> returnData = engineSocket.getRecieveBufferData();
                        int dataSize = returnData.size();
                        byte[] baData = new byte[dataSize];
                        for (int i = 0; i < dataSize; ++i) {
                            baData[i] = returnData.remove();
                        }
                        clientOutputStream.write(baData);
			TimingPill tp = new TimingPill(bytesToClient.addAndGet(dataSize));
			if(!sendTimes.offer(tp)){
			    sendTimes.poll();
			    if(!sendTimes.offer(tp)){
				System.out.println("[WARNING] Unable to add Timing Pill to sendTimes bounded queue. sendTimes.poll() did not help. ClientServiceRequest");
			    }
			}
			forwarder.totalBytesToClient.addAndGet(dataSize);

			// NOTE: the Save Recieve Data is from the Engine's perspective.
			if(ForwardProxy.saveReceiveData){
                            for(int i =0;i<baData.length;++i){
                                //Ticket 6.  Remove all non-ascii chars from the printed screen.
                                //           Define non-ascii as chars less than ascii 32 ' '.
                                if(baData[i]<32 && baData[i]!=10 & baData[i]!=13){
                                    baData[i]=126;
                                }
                            }
			    if(sentData.length()>1024000){
				//  We will truncate the sentData to protect memeory.
				//  TODO: save this off to file first.
				sentData.setLength(0);
			    }
			    sentData.append(new String(baData));
			}

                    } catch (IOException ex) {
                        //  If we ge an error here, we have to shut the system down.
                        shutdownReason.add("IOException Client Write");
			ex.printStackTrace(System.out);
                        shutdownSocketSystem();
                    }finally{
			dirty=true;
		    }

                }else{
		    // Check to see if the engine's communication has terminated.
		    // If so then we will do a single double check on the engine
		    // socket's recieve buffer before shutting down the request.
		    if(engineSocket.isEngineDone()){
			if(engineSocket.getRecieveBufferSize()==0){
			    try {
				// flush and close the socket to the Client.
				clientOutputStream.flush();
			    } catch (IOException ex) {
				// If we get an error here, there is nothing
				// we can do.  So we will log it and continue
				// to shut down the socket system.

				shutdownReason.add("[Exception] clientOutputStream.flush()");
				ex.printStackTrace(System.out);
			    }

			    //  This is the normal reason, all other
			    //  shutdowns should be considered as abnormal.
			    shutdownReason.add("Engine Done");

                            // Connections can
                            if(ForwardProxy.closeClientOnEngineClose){
                                shutdownSocketSystem();
                            }
			}
		    }
		    dirty=true;
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException ex) {
			ex.printStackTrace(System.out);
                    }
                }
            }
	    dirty=true;
        }
    }

    // This socket will write/send data down stream.
    public void run() {
	// 8 k buffer.
	byte[] buf = new byte[1024*8];
        while (isAlive) {

	    int bytesRead =0;
            try {
                if ((bytesRead=clientInputStream.read(buf))>-1) {
                    lastByteSent = System.currentTimeMillis();
		    //
		    TimingPill tp = new TimingPill(bytesFromClient.addAndGet(bytesRead));
		    if(!recieveTimes.offer(tp)){
			recieveTimes.poll();
			if(!recieveTimes.offer(tp)){
			    // TODO: Log this and find all similar ones to log.
			    System.out.println("Unable to Offer after removal. for Recieve Times in Client Service Request");
			}
		    }
		    forwarder.totalBytesFromClient.addAndGet(bytesRead);
		    // For this to be production ready,
	    // This data would have to be dealt with as
		    // primitive arrays.  The object actions are for
		    // ease of coding while in the debug mode.
                    LinkedList<Byte> inBytes = new LinkedList<Byte>();
                    for (int i = 0; i < bytesRead; ++i) {
                        inBytes.add(new Byte(buf[i]));
                    }

                    engineSocket.addSend(inBytes);


		    if(ForwardProxy.saveSendData){
                        // If we are savingSendData, we will have to strip out non-ascii chars.
                        // the reaason for this is so that you don't get beeping or other
                        // control chars showing up on the computer screen.
                        // (Ticket 6)
                        char[] bufChars = new char[bytesRead];
                        for(int i =0;i<bytesRead;++i){
                            // Ignore all chars less than ascii space ' '.
                            if(buf[i]<32&& buf[i]!=10 & buf[i]!=13){
                                 bufChars[i]=(char) 126;
                            }else{
                                bufChars[i]=(char)buf[i];
                            }
                        }

			if(recieveData.length()>1024000){
			    // For MEMORY PROTECTION.
			    // TODO: save this before truncate.
			    recieveData.setLength(0);
			}
			recieveData.append(bufChars);
		    }

                }else{
		    //  The input stream from the client has closed.
		    //  This will force a close to the entire duplex socket.
		    shutdownReason.add("End Of Stream -> clientInputStream");
		    shutdownSocketSystem();
		}
            } catch (IOException ex) {
               //  We will need to shut the socket down at this point.
		shutdownReason.add("IOException -> clientInputStream");
                shutdownSocketSystem();
            }finally{
		dirty=true;
	    }

        }
	shutdownReason.add("clientInputStream isAlive false");
	dirty=true;
    }

    private void shutdownSocketSystem(){
	try{
	    forwarder.removeServiceRequest(this);
	}catch(Exception e){
	    e.printStackTrace(System.out);
	}

	isAlive = false;
        //  Here we have to shut down all the socket activity.
        engineSocket.disconnectEngine();
        try {
            if(clientInputStream!=null){
                clientInputStream.close();
                clientInputStream =null;
            }
        } catch (IOException ex) {
            ex.printStackTrace(System.out);
        }

        try {
            if(clientOutputStream!=null){
		clientOutputStream.flush();
                clientOutputStream.close();
                clientOutputStream = null;
            }
        } catch (IOException ex) {
            ex.printStackTrace(System.out);
        }
        try {
            if(clientSocket!=null && clientSocket.isClosed()){
                clientSocket.close();
                clientSocket =null;
            }
        } catch (IOException ex) {
            ex.printStackTrace(System.out);
        }

        
	try{
	    System.out.println("[ShutdownSocketSystem: ] "+this.toString());
	}catch(Exception e){
	    e.printStackTrace(System.out);
	}

	dirty=true;
	finalTimer = true;
	endTime = System.currentTimeMillis();
    }

    /**
     *
		case 0: //  service Id
		case 1: //  duration
		case 2: //  start Time
		case 3: //  bytes from client
		case 4: //  bytes sent to engine
		case 5: //  bytes from engine
		case 6: //  bytes to client

     * @return
     */
    
    public Object[] currentState (){

	Object[] oa = new Object[11];
	oa[0]=  serviceId;
	oa[1] = (endTime<startTime) ? new Long(System.currentTimeMillis()-startTime): new Long(endTime-startTime);
	oa[2] = new Long(startTime);
	oa[3] = new Integer(bytesFromClient.get());
	oa[4] = new Integer(engineSocket.bytesToEngine.get());
	oa[5] = new Integer(engineSocket.bytesFromEngine.get());
	oa[6] = new Integer(bytesToClient.get());
        oa[7] = new Integer(clientLocalPort);//new Integer(-1);//"Port Eng";
        oa[8] = new Integer(engineSocket.getLocalPort());//new Integer(-1);//"Port Clnt";
        oa[9] = new Long(lastByteSent);// Pass back a long.  new Date(0);//"Clnt In Tm";
        oa[10]= new Long(engineSocket.getLastByteReceived());// Pass back a long.  new Date(0);//"Eng In Tm";

	return oa;

    }


    public int getLocalPort(){
        return clientLocalPort;
    }
    @Override
    public String toString(){
	StringBuilder sb = new StringBuilder();
	sb.append("ServiceId:").append(serviceId).append("\n");
	sb.append("CLIENT LOCAL PORT:").append(clientLocalPort).append("\n");
	sb.append("StartTime:").append(new Date(startTime)).append("\n");
	sb.append("Durration").append((endTime<startTime) ? new Long(System.currentTimeMillis()-startTime): new Long(endTime-startTime)).append("\n");
	sb.append("Bytes From Cleint:").append(bytesFromClient.get()).append("\n");
	sb.append("Bytes To Client:").append(bytesToClient.get()).append("\n");
	sb.append("DataSent:").append(sentData).append("\n");
	sb.append("DataRecieved:").append(recieveData).append("\n");
	sb.append("Send Timing Pills:\n");
	for (TimingPill timingPill : sendTimes) {
	    sb.append("\t").append(timingPill.toString()).append("\n");
	}
	sb.append("Recieve Timing Pills:\n");
	for (TimingPill timingPill : recieveTimes) {
	    sb.append("\t").append(timingPill.toString()).append("\n");
	}
	sb.append("Socket ShutDown Reason:\n");
	for (String reason : shutdownReason) {
	    sb.append("\t").append(reason).append("\n");
	}
	sb.append("Engine Socket STATS: \n").append(engineSocket.toString());

	return sb.toString();

    }
    public String getForwarderName(){
	return forwarder.toString();
    }


    @Override
    public int hashCode() {
	int hash = 3;
	hash = 53 * hash + (this.serviceId != null ? this.serviceId.hashCode() : 0);
	return hash;
    }

    @Override
    public boolean equals(Object obj) {
	if (obj == null) {
	    return false;
	}
	if (getClass() != obj.getClass()) {
	    return false;
	}
	final ClientServiceRequest other = (ClientServiceRequest) obj;
	if (this.serviceId != other.serviceId && (this.serviceId == null || !this.serviceId.equals(other.serviceId))) {
	    return false;
	}
	return true;
    }


}
